package com.lagou.wwyan.rpc.client;

import com.lagou.wwyan.rpc.bean.RpcRequestMonitor;
import com.lagou.wwyan.rpc.bean.RpcRequest;
import com.lagou.wwyan.rpc.bean.RpcResponse;
import com.lagou.wwyan.rpc.listener.ServiceChangeListener;
import com.lagou.wwyan.rpc.register.RpcRegister;

import java.lang.reflect.InvocationHandler;
import java.lang.reflect.Method;
import java.lang.reflect.Proxy;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;

public class RpcConsumer implements ServiceChangeListener {
    private Map<String, List<RpcClient>> CLIENT_POOL = new ConcurrentHashMap<>();
    private RpcRegister rpcRegister;
    private Map<String,Object> serviceMap;
//    private RpcClientSelector rpcClientSelector = new DefaultRpcClientSelector();
    private RpcClientSelector rpcClientSelector = new ZkRpcClientSelector();

    public RpcConsumer(RpcRegister rpcRegister, Map<String, Object> serviceMap) {
        this.rpcRegister = rpcRegister;
        this.serviceMap = serviceMap;

        for (Map.Entry<String, Object> entry : serviceMap.entrySet()) {
            String service = entry.getKey();
            List<String> servers = rpcRegister.discovery(service);
            List<RpcClient> rpcClients = CLIENT_POOL.computeIfAbsent(service, s -> new ArrayList<>());
            if (servers != null) {
                for (String server : servers) {
                    String[] split = server.split(":");
                    int port = Integer.parseInt(split[1]);
                    String host = split[0];
                    RpcRequestMonitor.add(host,port);
                    RpcClient client = new RpcClient(host, port);
                    try {
                        client.initClient();
                        rpcClients.add(client);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }

                }
            }
        }
        this.rpcRegister.addListener(this);
    }

    @Override
    public void serverAdd(String service, String host, int port) {
        List<RpcClient> clients = CLIENT_POOL.computeIfAbsent(service,s -> new ArrayList<>());
        RpcRequestMonitor.add(host,port);
        RpcClient client = new RpcClient(host, port);
        try {
            client.initClient();
            clients.add(client);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    @Override
    public void serverRemove(String service, String host, int port) {
        List<RpcClient> clients = CLIENT_POOL.get(service);
        if (clients == null || clients.isEmpty()){
            return;
        }
        clients.removeIf(next -> Objects.equals(next.getHost(), host) && port == next.getPort());
        RpcRequestMonitor.remove(host,port);
    }
    @SuppressWarnings("unchecked")
    public <T>  T createProxy(Class<T> clazz){
        return (T) Proxy.newProxyInstance(Thread.currentThread().getContextClassLoader(), new Class[]{clazz},new InvocationHandler() {
            public Object invoke(Object o, Method method, Object[] args) throws Throwable  {
                RpcRequest rpcRequest = new RpcRequest();
                rpcRequest.setRequestId(UUID.randomUUID().toString());
                String service = clazz.getName();
                rpcRequest.setClassName(service);
                rpcRequest.setMethodName(method.getName());
                rpcRequest.setParameters(args);
                if (args != null) {
                    Class<?>[] parameterTypes = new Class[args.length];
                    for (int i = 0; i < args.length; i++) {
                        Class<?> aClass = args[i].getClass();
                        parameterTypes[i] = aClass;
                    }
                    rpcRequest.setParameterTypes(parameterTypes);
                } else {
                    rpcRequest.setParameterTypes(new Class[0]);
                }
                RpcClient client = rpcClientSelector.select(service, CLIENT_POOL);
                if (client != null){
                    RpcResponse response = (RpcResponse)client.send(rpcRequest);
                    return response.getData();
                }
                return "error";
//            return "success";
            }
        });
    }
}
